package com.zhj.retrofitwithrxjavademo;

import android.app.Application;
import android.test.ApplicationTestCase;

import com.alpha.demoretrofit.MainActivity;

/**
 * <a href="http://d.android.com/tools/testing/testing_android.html">Testing Fundamentals</a>
 */
public class ApplicationTest extends ApplicationTestCase<Application> {
    private static final String TAG = MainActivity.class.getSimpleName();

    public ApplicationTest() {
        super(Application.class);

        /*
        * 上游可以发送无限个onNext, 下游也可以接收无限个onNext.
            当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
            当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
            上游可以不发送onComplete或onError.
            最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete, 然后再发一个onError, 反之亦然
        *
        *
        * */

//        /*多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.*/
//        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                Log.d(TAG, "Observable thread is : " + Thread.currentThread().getName());
//                Log.d(TAG, "emit 1");
//                emitter.onNext(1);
//            }
//        });
//
//        Consumer<Integer> consumer = new Consumer<Integer>() {
//            @Override
//            public void accept(Integer integer) throws Exception {
//                Log.d(TAG, "Observer thread is :" + Thread.currentThread().getName());
//                Log.d(TAG, "onNext: " + integer);
//            }
//        };
//
//        //多次指定上游的线程只有第一次指定的有效, 也就是说多次调用subscribeOn() 只有第一次的有效, 其余的会被忽略.
//        observable.subscribeOn(Schedulers.newThread())
//                .observeOn(AndroidSchedulers.mainThread()) //将结果放到主线程当中
//                .subscribe(consumer);


//        Observable.create(new ObservableOnSubscribe<Integer>() {
//
//            // 只有当上游和下游建立连接之后, 上游才会开始发送事件.
//            // 也就是调用了subscribe() 方法之后才开始发送事件.
//            @Override
//            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//                e.onNext(1);//发送信息
//                e.onNext(2);
//                e.onNext(3);
//                e.onComplete();//发送完成
//            }
//        }).flatMap(new Function<Integer, ObservableSource<String>>() {
//            @Override
//            public ObservableSource<String> apply(Integer integer) throws Exception {
//                final List<String> list = new ArrayList<>();
//                for (int i = 0; i < 3; i++) {
//                    list.add("I am value " + integer);
//                }
//                return Observable.fromIterable(list).delay(10, TimeUnit.MILLISECONDS);
//            }
//        }).subscribe(new Observer<String>() {
//            private Disposable mDisposable;
//            private int i;
//
//            @Override
//            public void onSubscribe(Disposable d) {
//                Log.d(TAG, "subscribe");
//                mDisposable = d;
//            }
//
//            @Override
//            public void onNext(String value) {
//                Log.d(TAG, "onNext：" + value);
//                i++;
//                if (i == 2) {
//                    // 在收到onNext 2这个事件后, 切断了水管, 但是上游仍然发送了3, complete,
//                    Log.d(TAG, "dispose");
//                    mDisposable.dispose();
//                    Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
//                }
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.d(TAG, "onError");
//            }
//
//            @Override
//            public void onComplete() {
//                Log.d(TAG, "onComplete");
//            }
//
//
//        });

//
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//
//            // 只有当上游和下游建立连接之后, 上游才会开始发送事件.
//            // 也就是调用了subscribe() 方法之后才开始发送事件.
//            @Override
//            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//                e.onNext(1);//发送信息
//                e.onNext(2);
//                e.onNext(3);
//                e.onComplete();//发送完成
//            }
//        })
        //不带任何参数的subscribe() 表示下游不关心任何事件,你上游尽管发你的数据去吧, 老子可不管你发什么.
//        带有一个Consumer参数的方法表示下游只关心onNext事件, 其他的事件我假装没看见, 因此我们如果只需要onNext事件可以这么写:
//                .subscribe(new Consumer<Integer>() {
//                    @Override
//                    public void accept(Integer integer) throws Exception {
//
//                    }
//                })
//
//                .subscribe(new Observer<Integer>() {
//                    private Disposable mDisposable;
//                    private int i;
//
//                    @Override
//                    public void onSubscribe(Disposable d) {
//                        Log.d(TAG, "subscribe");
//                        mDisposable = d;
//                    }
//
//                    @Override
//                    public void onNext(Integer value) {
//                        Log.d(TAG, "onNext：" + value);
//                        i++;
//                        if (i == 2) {
//                            // 在收到onNext 2这个事件后, 切断了水管, 但是上游仍然发送了3, complete,
//                            Log.d(TAG, "dispose");
//                            mDisposable.dispose();
//                            Log.d(TAG, "isDisposed : " + mDisposable.isDisposed());
//                        }
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        Log.d(TAG, "onError");
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        Log.d(TAG, "onComplete");
//                    }
//
//
//                });
//        /*zip操作*/
//        Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
//
//            @Override
//            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
//                e.onNext(1);
//                e.onNext(2);
//                e.onNext(3);
//            }
//        }).subscribeOn(Schedulers.io());
//
//        Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
//
//            @Override
//            public void subscribe(ObservableEmitter<String> e) throws Exception {
//                e.onNext("A");
//                e.onNext("B");
//                e.onNext("C");
//            }
//        }).subscribeOn(Schedulers.io());
//
//        Observable.zip(integerObservable, stringObservable, new BiFunction<Integer, String, String>() {
//            @Override
//            public String apply(Integer integer, String s) throws Exception {
//                return integer + s;
//            }
//        }).subscribe(new Observer<String>() {
//            @Override
//            public void onSubscribe(Disposable d) {
//
//            }
//
//            @Override
//            public void onNext(String value) {
//                Log.e(TAG, "合并之后的值：" + value);
//            }
//
//            @Override
//            public void onError(Throwable e) {
//
//            }
//
//            @Override
//            public void onComplete() {
//
//            }
//        });

         /*分流处理*/
//        Observable.create(new ObservableOnSubscribe<Integer>() {
//            @Override
//            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
//                for (int i = 0; ; i++) {
//                    emitter.onNext(i);
//                }
//            }
//        }).subscribeOn(Schedulers.io())
//                .filter(new Predicate<Integer>() {
//                    @Override
//                    public boolean test(Integer integer) throws Exception {
//                        return integer % 10 == 0;
//                    }
//                })
//                .observeOn(AndroidSchedulers.mainThread())
//                .subscribe(new Consumer<Integer>() {
//                    @Override
//                    public void accept(Integer integer) throws Exception {
//                        Log.d(TAG, "" + integer);
//                    }
//                });

//        Flowable.create(new FlowableOnSubscribe<String>() {
//
//            @Override
//            public void subscribe(FlowableEmitter<String> e) throws Exception {
//
//            }
//        }, BackpressureStrategy.ERROR);
    }
}